package com.pai4j.common.service.messagequeue.consumer;

import com.pai4j.common.enums.messagequeue.MessageQueueEnum;
import com.pai4j.common.util.RedisUtil;
import com.pai4j.domain.dto.queue.BaseDelayQueueDTO;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;

import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * 抽象的redis延迟队列消费者的实现方法
 *
 * @author: CYM-pai
 * @date: 2025/04/29 21:14
 **/
@Slf4j
public abstract class AbstractRedisDelayQueueConsumer<T extends BaseDelayQueueDTO> {

    @Resource
    protected RedisUtil redisUtil;

    // 标记是否已初始化的volatile布尔变量，保证多线程环境下的可见性
    //volatile确保多线程环境下的可见性，即当一个线程修改了isInit的值，其他线程能够立即看到最新的值
    private volatile boolean isInit = false;
    
    // 标记是否继续运行的volatile布尔变量，用于控制线程的执行状态
    private volatile boolean isRun = true;
    
    // 使用线程池执行任务，提高线程的复用性和效率
    private ThreadPoolExecutor poolExecutor;
    /**
     * 初始化消息队列及其相关资源
     * 此方法首先检查消息队列是否已经初始化，如果已经初始化，则直接返回
     * 否则，它将开始初始化过程，包括日志记录、线程池初始化以及状态更新
     */
    public void init() {
        // 检查消息队列是否已经初始化
        if (isInit) {
            return;
        }
        
        // 记录初始化消息队列的开始信息和配置参数
        log.info("开始初始化消息队列，参数如下: \n" +
                        "queue name: {} \n" +
                        "queue timeout: {}\n" +
                        "queue retry times: {}\n" +
                        "queue back queue times: {}\n" +
                        "queue pool size: {}",
                getMessageQueue().getQueue(), getTimeout(), getRetryTimes(), getBackQueueTimes(), getPoolSize()
        );
        
        // 根据配置的线程池大小初始化线程池
        initThreadPool(getPoolSize());
        // 获取并记录消息队列启动时的元素数量
        Long startSize = redisUtil.zSetSize(getQueueName());
        log.info("启动消息队列{}成功, 目前还有 {} 个元素没被消费", getQueueName(), startSize);
        
        // 更新初始化状态
        isInit = true;
    }

    /**
     * 从消息队列redis读取数据的延迟
     *
     * @return
     */
    protected long getTimeout() {
        return 3000;
    }

    /**
     * 一个任务元素重试的次数
     *
     * @return 默认不重试
     */
    protected int getRetryTimes() {
        return 0;
    }

    /**
     * 一个任务最多被重返队列的次数
     *
     * @return
     */
    protected int getBackQueueTimes() {
        
        return 0;
    }

    /**
     * 获取最大执行任务数量
     *
     * @return
     */
    protected int getPoolSize() {
        return 1;
    }

    /**
     * 将字符串转化为对象
     *
     * @param s
     * @return
     */
    protected abstract T convert(String s);

    /**
     * 队列信息
     *
     * @return
     */
    protected abstract MessageQueueEnum getMessageQueue();

    /**
     * 执行下一步消费实际处理逻辑
     */
    protected abstract void doNextConsume(Set<String> messages);

    protected String getQueueName() {
        
        return this.getMessageQueue().getQueue();
    }

    /**
     * 延迟队列消费
     *
     * 从延迟队列中消费消息，即获取并删除队列中所有score小于等于当前时间的消息
     * 此方法解释了为什么需要根据当前时间获取消息以及为什么获取后需要删除的原因
     *
     * @return 消费的消息集合，如果队列中没有可消费的消息则返回空集合
     */
    private Set<String> consumeMessage() {
        // 当前时间
        long currentTime = System.currentTimeMillis();
        // 按照score升序取消息任务:返回的数据按score从小到大排序，所以得到的messages:["8:00","8:01","8:02","8:03","9:01","9:01"]
        Set<String> messages = redisUtil.zrangeByScore(getQueueName(), Long.MIN_VALUE, currentTime);
        if (CollectionUtils.isNotEmpty(messages)) {
            // 消费到了数据，需要删除队列中数据，否则会重复消费
            redisUtil.zremoveRangeByScore(getQueueName(), Long.MIN_VALUE, currentTime);
        }
        return messages;
    }

    /**
     * 负责调度的核心方法
     * 通过循环读取redis的阻塞队列，获取元素，并且转化为用户可直接操作的类型
     * 如果处理失败采用两种容错方式:
     * 1. 重试，可以通过设置retryTime来设置重试次数
     * 2. 重新归队，只有在重试全部失败后才回执行该逻辑，可以设置backQueueTimes设置归队的最大次数
     * 如果两种方式都失败，则进入用户自定义的容错方式中
     */
    public void schedule() {
        init();
        try {
            //通过信号量控制并发执行的任务数量，确保同时运行的线程不超过设定的池大小
            final Semaphore semaphore = new Semaphore(getPoolSize());
            while (isRun) {
                Thread.sleep(1000);
                // 判断是否目前达到了使用上限
                log.info("QueueName:{} 信号量剩余:{}", getQueueName(), semaphore.availablePermits());
                semaphore.acquire();
                // 获取需要消费的消息并从队列中删除，防止重复消费
                final Set<String> messages = this.consumeMessage();
                if (CollectionUtils.isEmpty(messages)) {
                    log.info("QueueName:{} 目前没有元素", getQueueName());
                    semaphore.release();
                    continue;
                }
                poolExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            //具体的消息消费处理
                            doNextConsume(messages);
                        } finally {
                            // 任务执行完释放信号量
                            semaphore.release();
                        }
                    }
                });
            }
        } catch (final Exception outerEx) {
            log.error("启动消息队列失败, 原因 ", outerEx);
        }
    }

    /**
     * 初始化数据库连接池
     *
     * @param poolSize 连接池的大小，即最大线程数
     */
    private void initThreadPool(int poolSize) {
        // 双重检查锁定，确保线程安全地初始化线程池
        if (poolExecutor == null) {
            synchronized (this) {
                // 再次检查线程池是否已经初始化，以避免多次初始化
                if (poolExecutor == null) {
                    // 使用固定大小的线程池，以便更好地控制资源使用
                    poolExecutor = (ThreadPoolExecutor) Executors.newFixedThreadPool(poolSize);
                }
            }
        }
    }
}
